package com.xsomnus.rpc.common.config;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.ACLProvider;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheListener;
import org.apache.curator.framework.recipes.locks.InterProcessReadWriteLock;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Stat;

import java.nio.charset.Charset;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;

/**
 * @author @xsomnus666_xiawenye★
 * @since 2019/4/18 0018 10:56
 * 当你发现自己的才华撑不起野心时，就请安静下来学习吧
 */
@Data
@Slf4j
public class ZkClient {

    private CuratorFramework client;

    private ZookeeperProperties zookeeperProperties;

    private TreeCache treeCache;

    public ZkClient(ZookeeperProperties zookeeperProperties) {
        this.zookeeperProperties = zookeeperProperties;
    }

    public void init() {
        try {
            RetryPolicy retryPolicy = new ExponentialBackoffRetry(zookeeperProperties.getBaseSleepTimeMs(), zookeeperProperties.getMaxRetries());
            CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder()
                    .connectString(zookeeperProperties.getServer())
                    .retryPolicy(retryPolicy)
                    .sessionTimeoutMs(zookeeperProperties.getSessionTimeoutMs())
                    .connectionTimeoutMs(zookeeperProperties.getConnectionTimeoutMs())
                    .namespace(zookeeperProperties.getNamespace());
            if (StringUtils.isNoneEmpty(zookeeperProperties.getDigest())) {
                builder.authorization("digest", zookeeperProperties.getDigest().getBytes(Charset.forName("UTF-8")));
                builder.aclProvider(new ACLProvider() {
                    @Override
                    public List<ACL> getDefaultAcl() {
                        return ZooDefs.Ids.CREATOR_ALL_ACL;
                    }

                    @Override
                    public List<ACL> getAclForPath(final String path) {
                        return ZooDefs.Ids.CREATOR_ALL_ACL;
                    }
                });
            }

            client = builder.build();
            client.start();
            initLocalCache("/test");

            client.getConnectionStateListenable().addListener((curatorFramework, connectionState) -> {
                log.debug("curatorFramework: {}, connectionState: {}", curatorFramework, connectionState);
                if (connectionState == ConnectionState.LOST) {
                    log.debug("zookeeper连接丢失");
                } else if (connectionState == ConnectionState.CONNECTED) {
                    log.debug("zookeeper连接成功建立");
                } else if (connectionState == ConnectionState.RECONNECTED) {
                    log.debug("zookeeper重新连接");
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private void initLocalCache(final String watchRootPath) throws Exception {
        treeCache = new TreeCache(client, watchRootPath);
        TreeCacheListener listener = (client, event) -> {
            log.debug("event-type: {}, | path = {}", event.getType(), (null != event.getData()) ? event.getData().getPath() : null);
            if (event.getData() != null && event.getData().getData() != null) {
                log.debug("发生变化的内容的节点为:{}", new String(event.getData().getData()));
            }
        };
        treeCache.getListenable().addListener(listener);
        treeCache.start();
    }

    //---------------------------------------------------------------------
    //  CURD API
    //---------------------------------------------------------------------

    /**
     * 创建节点
     *
     * @param mode 节点类型
     *             1、PERSISTENT 持久化目录节点，存储的数据不会丢失。
     *             2、PERSISTENT_SEQUENTIAL顺序自动编号的持久化目录节点，存储的数据不会丢失
     *             3、EPHEMERAL临时目录节点，一旦创建这个节点的客户端与服务器端口也就是session 超时，这种节点会被自动删除
     *             4、EPHEMERAL_SEQUENTIAL临时自动编号节点，一旦创建这个节点的客户端与服务器端口也就是session 超时，这种节点会被自动删除，并且根据当前已近存在的节点数自动加 1，然后返回给客户端已经成功创建的目录节点名。
     * @param path 节点名称
     */
    public void createNode(CreateMode mode, String path, String nodeData) {
        try {
            String ret = client.create().creatingParentsIfNeeded().withMode(mode).forPath(path, nodeData.getBytes(Charset.forName("UTF-8")));
            log.debug("添加成功之后的返回值: {}", ret);
        } catch (Exception e) {
            e.printStackTrace();
            log.error("添加节点错误,errMsg = {}", e.getLocalizedMessage());
        }
    }

    public void createNode(CreateMode mode, String path) {
        try {
            String ret = client.create().creatingParentsIfNeeded().withMode(mode).forPath(path);
            log.debug("添加成功之后的返回值: {}", ret);
        } catch (Exception e) {
            e.printStackTrace();
            log.error("添加节点错误,errMsg = {}", e.getLocalizedMessage());
        }
    }


    public void setNodeData(String path, byte[] data) {
        try {
            Stat ret = client.setData().forPath(path, data);
            log.debug("更新节点之后的返回值:{}", ret);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public byte[] getNodeData(String path) {
        try {
            if (treeCache != null) {
                ChildData currentData = treeCache.getCurrentData(path);
                if (currentData != null) {
                    return currentData.getData();
                }
            }
            return client.getData().forPath(path);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    public byte[] syncNodeData(String path) {
        client.sync();
        return getNodeData(path);
    }

    public boolean checkExists(final String path) {
        client.sync();
        try {
            return null != client.checkExists().forPath(path);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    public List<String> listChildren(String path) {
        try {
            return client.getChildren().forPath(path);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return Collections.emptyList();
    }

    public void deleteNode(final String path) {
        try {
            client.delete().guaranteed().deletingChildrenIfNeeded().forPath(path);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void deleteNode(final String path, boolean deleteChildren) {
        try {
            if (deleteChildren) {
                client.delete().guaranteed().deletingChildrenIfNeeded().forPath(path);
            } else {
                client.delete().guaranteed().forPath(path);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 随机读取一个path子路径, "/"为根节点对应该namespace
     * 先从cache中读取，如果没有，再从zookeeper中查询
     *
     * @param path 路径
     */
    public String randomNodeData(String path) {
        Map<String, ChildData> cacheMap = treeCache.getCurrentChildren(path);
        if (null != cacheMap && cacheMap.size() > 0) {
            log.debug("从节点[{}]随机获取值", path);
            byte[] data = cacheMap.values().stream().findAny().orElseThrow(() -> new RuntimeException("一个都没有吗?")).getData();
            return new String(data, Charset.forName("UTF-8"));
        }

        if (checkExists(path)) {
            log.debug("path [{}] 为空", path);
            return null;
        } else {
            try {
                List<String> children = client.getChildren().forPath(path);
                if (children == null || children.size() == 0) {
                    log.debug("path [{}] 没有子节点", path);
                    return null;
                }
                String dest = children.stream().findAny().get();
                path = path + "/" + dest;
                byte[] data = client.getData().forPath(path);
                return new String(data, Charset.forName("UTF-8"));
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        return null;
    }

    //---------------------------------------------------------------------
    //  功能API
    //---------------------------------------------------------------------

    /**
     * 获取读写锁
     * @param path 节点路径
     */
    public InterProcessReadWriteLock getReadWriteLock(String path) {
        return new InterProcessReadWriteLock(client, path);
    }


    /**
     * 监听节点数据的变化
     */

    public void watchPath(String watchPath, TreeCacheListener listener) {
        TreeCache cache  = new TreeCache(client, watchPath);
        cache.getListenable().addListener(listener, Executors.newFixedThreadPool(3));
        try {
            cache.start();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    public List<String> watchPathV2(String watchPath, TreeCacheListener listener) {
        TreeCache cache  = new TreeCache(client, watchPath);
        cache.getListenable().addListener(listener, Executors.newFixedThreadPool(3));
        try {
            cache.start();
            return client.getChildren().forPath(watchPath);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return Collections.emptyList();
    }


    public void stop() {
        if (client != null) {
            client.close();
        }
    }
}
